package com.gxlevi.batch_process

import com.gxlevi.batch_process.bean.{OrderRecord, OrderRecordWide}
import com.gxlevi.batch_process.task.{MerchantCountMoneyTask, PayMethodMoneyCountTask, PreprocessTask}
import com.gxlevi.batch_process.util.HBaseTableInputFormat
import org.apache.flink.api.java.tuple
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._

object App {
  def main(args: Array[String]): Unit = {
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tupleDataSet: DataSet[tuple.Tuple2[String, String]] = env.createInput(new HBaseTableInputFormat("mysql.gxlevi.orderRecord"))

    val orderRecorderDataSet: DataSet[OrderRecord] = tupleDataSet.map {
      tuple2 => OrderRecord(tuple2.f1)
    }

    val orderRecordDataSet: DataSet[OrderRecordWide] = PreprocessTask.process(orderRecorderDataSet)
    //按照天、月、年的维度来统计出不同支付方式的订单数量、支付金额
    //PayMethodMoneyCountTask.process(orderRecordDataSet)
    //根据不同商家，按照天、月、年、所有维度来进行统计订单数量、支付金额
    MerchantCountMoneyTask.process(orderRecordDataSet)

  }
}
